package org.joa.actor;

import java.io.Serializable;
import java.util.UUID;
import java.util.concurrent.Callable;

import org.joa.actor.exception.RemoteException;
import org.joa.actor.transaction.Transaction;
import org.joa.actor.transaction.TransactionProxy;
import org.joa.manager.ActorManager;
import org.joa.manager.ManagedActor;
import org.joa.manager.exception.TransactionLockTimeoutException;

public class Actor implements Serializable {
    private static final long serialVersionUID = 1L;

    private ActorName name;
    private transient ActorManager actorManager = ActorManager.instance;

    public Actor() {
        this(new ActorName(UUID.randomUUID()));
    }

    public Actor(ActorName name) {
        this.name = name;
    }

    public ActorName self() {
        return name;
    }

    public Transaction executeAtomic(final Runnable task) throws RemoteException {
        return executeAtomic(new Callable<Void>() {
            @Override
            public Void call() throws Exception {
                task.run();
                return null;
            }
        });
    }

    public Transaction executeAtomic(Callable<?> task) throws RemoteException {
        ManagedActor managedActor = actorManager.getActor(self());

        Transaction transaction = null;
        synchronized (managedActor) {
            transaction = getCurrentTransaction();

            if (transaction != null) {
                // We are already in a previously created transaction
                try {
                    task.call();
                } catch (Exception e) {
                    throw new RemoteException(e);
                }
                return transaction;
            }

            transaction = TransactionProxy.create(self(), task);
            managedActor.setCurrentTransaction(transaction);
        }

        transaction.incrementPendingMessageCount(self());

        try {
            task.call();
        } catch (TransactionLockTimeoutException ex) {
            // The call attempt failed due to lock timeout
        } catch (Exception e) {
            transaction.rollback();
            throw new RuntimeException(e);
        } finally {
            transaction.decrementPendingMessageCount(self());
        }

        return transaction;
    }

    public Transaction getCurrentTransaction() {
        ManagedActor managedActor = actorManager.getActor(self());
        return managedActor.getCurrentTransaction();
    }

    public String toString() {
        return name.toString();
    }

    public boolean equals(Object otherObject) {
        if (!(otherObject instanceof Actor)) {
            return false;
        }

        Actor otherActor = (Actor) otherObject;
        return self().equals(otherActor.self());
    }

    public int hashCode() {
        return self().hashCode();
    }
}
